package com.aispeech.asr.webase.component;

import com.fasterxml.jackson.databind.JsonNode;
import io.netty.channel.ChannelOption;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.client.MultipartBodyBuilder;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.*;
import java.time.Duration;
import java.util.concurrent.*;

@Component
@Slf4j
@Lazy
public class BigFileHttpPoster {
    @Value("${apis.audio-upload.url}")
    private String url;
    @Value("${apis.audio-upload.collection}")
    private String collection;
    @Value("${apis.audio-upload.apikey}")
    private String apiKey;
    @Value("${apis.audio-upload.secret}")
    private String secret;

    private ThreadPoolExecutor executor;

    @PostConstruct
    private void init() {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(1000);
        executor = new ThreadPoolExecutor(10, 50, 0L, TimeUnit.MILLISECONDS, queue);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());

        log.info("[ BigFileHttpPoster ] init");
    }

    public void run() {
    }

    private WebClient createAndInitWebClient() {
        HttpClient httpClient = HttpClient.create()
                .tcpConfiguration(tcpClient -> tcpClient
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                        .doOnConnected(connection -> {
                            connection
                                    .addHandlerLast(new ReadTimeoutHandler(60))
                                    .addHandlerLast(new WriteTimeoutHandler(60));
                        }));

        WebClient webClient = WebClient
                .builder()
                .baseUrl(url)
                .filter(ExchangeFilterFunctions.basicAuthentication(apiKey, secret))
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();

        return webClient;
    }

    /*
     * 加入要异步上传的文件。
     * filePath： 文件的绝对路径。
     * objectKey： 上传成功之后，下载指定的file_name。
     */
    public void add(String filePath, String objectKey, ByteArrayResource resource) {
        BigFile bigFile = new BigFile(this, filePath, objectKey, resource);

        try {
            executor.submit(bigFile);
        } catch (RejectedExecutionException ex) {
            log.error("[ BigFileHttpPoster.add ] RejectedExecutionException occurred. backupToLocal: {}", filePath);
            backupToLocal(bigFile);
        }
    }

    public void doUpload(BigFile bigFile) {
        MultipartBodyBuilder builder = new MultipartBodyBuilder();

        builder.part("objectKey", bigFile.objectKey);
        builder.part("collection", collection);
        builder.part("file", bigFile.resource).filename(bigFile.objectKey).contentType(MediaType.APPLICATION_OCTET_STREAM);

        try {
            JsonNode json = createAndInitWebClient()
                    .post()
                    .contentType(MediaType.MULTIPART_FORM_DATA)
                    .body(BodyInserters.fromMultipartData(builder.build()))
                    .retrieve()
                    .onStatus(HttpStatus::is4xxClientError, response -> backupToLocal(response, bigFile))
                    .onStatus(HttpStatus::is5xxServerError, response -> backupToLocal(response, bigFile))
                    .bodyToMono(JsonNode.class)
                    .retry(1)
                    .doOnError(throwable -> {
                        log.error("[ backupToLocal ] failed send file to server. backup file: {}.", bigFile.objectKey, throwable);
                        backupToLocal(bigFile);
                    })
                    .block(Duration.ofSeconds(30));

            log.info("[ doUpload ] resp: {}, objectKey: {}", json, bigFile.objectKey);
        } catch (Exception ex) {
            log.warn("[ doUpload ] backup file: {},ex:", bigFile.objectKey, ex);
            backupToLocal(bigFile);
        }
    }

    public Mono<WebClientResponseException> backupToLocal(ClientResponse response, BigFile bigFile) {
        OutputStream outputStream = null;
        String fileName = bigFile.filePath;
        Mono<WebClientResponseException> exceptionMono = response.createException();
        exceptionMono.map(ex -> {
            log.warn("[ backupToLocal ] failed send file to server. backup file: {}, response: {} (code: {}).", fileName,
                    ex.getResponseBodyAsString(), ex.getRawStatusCode());
            return ex;
        });

        try {
            File file = new File(fileName);
            if (!file.getParentFile().exists()) file.getParentFile().mkdir();

            InputStream inputStream = bigFile.resource.getInputStream();
            outputStream = new FileOutputStream(fileName);
            byte[] buff = new byte[8192];
            int len;

            while ((len = inputStream.read(buff)) != -1) {
                outputStream.write(buff, 0, len);
            }
        } catch (IOException e) {
            log.warn("backup file: {} failed.", fileName);
        } finally {
            if (outputStream != null) {
                try {
                    outputStream.close();
                } catch (IOException e) {
                }
            }
        }

        return exceptionMono;
    }

    public void backupToLocal(BigFile bigFile) {
        OutputStream outputStream = null;
        String fileName = bigFile.filePath;

        try {
            File file = new File(fileName);
            if (!file.getParentFile().exists()) file.getParentFile().mkdir();

            InputStream inputStream = bigFile.resource.getInputStream();
            outputStream = new FileOutputStream(fileName);
            byte[] buff = new byte[8192];
            int len;

            while ((len = inputStream.read(buff)) != -1) {
                outputStream.write(buff, 0, len);
            }
        } catch (IOException e) {
            log.warn("backup file: {} failed.", fileName);
        } finally {
            if (outputStream != null) {
                try {
                    outputStream.close();
                } catch (IOException e) {
                }
            }
        }
    }

    @PreDestroy
    private void finit() {
        executor.shutdown();
        log.info("[ BigFileHttpPoster ] finit");
    }

    class BigFile implements Runnable {
        public String objectKey;
        public String filePath;
        public Resource resource;
        public BigFileHttpPoster bigFileHttpPoster;

        BigFile(BigFileHttpPoster bigFileHttpPoster, String filePath, String objectKey, Resource resource) {
            this.bigFileHttpPoster = bigFileHttpPoster;
            this.filePath = filePath;
            this.objectKey = objectKey;
            this.resource = resource;
        }

        @Override
        public void run() {
            this.bigFileHttpPoster.doUpload(this);
        }
    }
}
